package com.qupeng.demo.kafka.kafkaspringbootproducer.admin;

import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Coordinator;
import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Partition;
import com.qupeng.demo.kafka.kafkaspringbootproducer.admin.model.Topic;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.TopicPartitionInfo;
import org.springframework.beans.BeanUtils;

public class Util {
    public static Topic convert(TopicDescription topicDescription) {
        Topic topic = new Topic();
        BeanUtils.copyProperties(topicDescription, topic);
        topicDescription.partitions().forEach(topicPartitionInfo -> {
            topic.getPartitions().add(Util.convert(topicPartitionInfo));
        });

        return topic;
    }

    public static Partition convert(TopicPartitionInfo partitionInfo) {
        Partition partition = new Partition();
        BeanUtils.copyProperties(partitionInfo, partition);
        BeanUtils.copyProperties(partitionInfo.leader(), partition.getLeader());
        partitionInfo.replicas().forEach(node -> {
            Coordinator replica = new Coordinator();
            BeanUtils.copyProperties(node, replica);
            partition.getReplicas().add(replica);
        });
        partitionInfo.isr().forEach(node -> {
            Coordinator replica = new Coordinator();
            BeanUtils.copyProperties(node, replica);
            partition.getIsr().add(replica);
        });

        return partition;
    }
}
